/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.route;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;
import com.pingan.haofang.searchcloud.api.route.IndexRoute;

/**
 * @author huhailiang937
 * @date 2017年7月6日下午12:02:54
 * @since 1.0.0
 */
public class RouteChangeZKListener {

    protected static Logger LOG = LoggerFactory.getLogger(RouteChangeZKListener.class);

    private String metaZkHost;

    protected CuratorFramework client = null;

    protected PathChildrenCache pathChildrenCache = null;

    public RouteChangeZKListener(String metaZkHost) {
        this.metaZkHost = metaZkHost;
    }

    public void start(final Cache<String, IndexRoute> registerCache) {
        LOG.info("-------------> MetadataChangeListener start");
        CuratorFramework client = CuratorFrameworkFactory.newClient(metaZkHost, new ExponentialBackoffRetry(10000, 3));
        client.start();

        pathChildrenCache = new PathChildrenCache(client, MetaCacheSupport.ROOT_ROUTE_PATH, true);
        try {
            pathChildrenCache.start();

            PathChildrenCacheListener listener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                    switch (event.getType()) {
                    case CHILD_ADDED: {
                        LOG.debug("--->Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Node added: " + new String(event.getData().getData()));
                            addOrUpdateCollectionRoute(event.getData().getData(), registerCache);
                        }

                        break;
                    }

                    case CHILD_UPDATED: {
                        LOG.debug("--->Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Node changed: " + new String(event.getData().getData()));
                            addOrUpdateCollectionRoute(event.getData().getData(), registerCache);
                        }
                        break;
                    }

                    case CHILD_REMOVED: {
                        LOG.debug("--->Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                        if (null != event.getData().getData()) {
                            LOG.debug("--->Node removed: " + new String(event.getData().getData()));
                            removeCollectionRoute(event.getData().getData(), registerCache);
                        }
                        break;
                    }
                    default:
                        break;
                    }
                }
            };
            pathChildrenCache.getListenable().addListener(listener);
        } catch (Exception e) {
            LOG.error("PathChildrenCache start has error:", e);
        }
    }

    public void destroy() {
        if (null != client) {
            CloseableUtils.closeQuietly(client);
        }
        if (null != pathChildrenCache) {
            CloseableUtils.closeQuietly(pathChildrenCache);
        }
    }

    private void addOrUpdateCollectionRoute(byte[] datas, Cache<String, IndexRoute> registerCache) {
        try {
            IndexRoute indexRoute = MetaCacheSupport.decode(datas, IndexRoute.class);
            registerCache.put(indexRoute.getCollectionRouteKey(), indexRoute);
        } catch (Exception e) {
            LOG.error("addOrUpdateCollectionRoute has error:", e);
        }

    }

    private void removeCollectionRoute(byte[] datas, Cache<String, IndexRoute> registerCache) {
        try {
            IndexRoute indexRoute = MetaCacheSupport.decode(datas, IndexRoute.class);
            registerCache.invalidate(indexRoute.getCollectionRouteKey());
        } catch (Exception e) {
            LOG.error("removeCollectionRoute has error:", e);
        }
    }

}
